package spark_core;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.ResultSet;
import java.sql.SQLException;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.rdd.JdbcRDD;
import scala.reflect.ClassManifestFactory;
import scala.runtime.AbstractFunction0;
import scala.runtime.AbstractFunction1;


/**
 * @author shihb
 * @date 2020/1/9 17:58
 */
public class JdbcRddDemo {

  public static void main(String[] args) throws Exception {


    //local模式,创建SparkConf对象设定spark的部署环境
    SparkConf sparkConf = new SparkConf().setMaster("local[*]").setAppName("test");
    //创建spark上下文对象（这边是java上下文）
    JavaSparkContext jsc = new JavaSparkContext(sparkConf);

    String driver = "com.mysql.jdbc.Driver";
    String url = "jdbc:mysql://linux1:3306/raa";
    String user = "root";
    String password = "123456";

    String sql = "select * form user where id >=? and id <= ?";
    //创建JdbcRdd,查询数据库

    JdbcRDD<String> jdbcRdd = new JdbcRDD<String>(
        jsc.sc(),
        new AbstractFunction0<Connection>() {
          @Override
          public Connection apply(){
            try {
              Class.forName(driver);
              Connection conn = DriverManager.getConnection(url, user, password);
              return conn;
            } catch (Exception e) {
              throw new RuntimeException("get jdbc connection error");
            }
          }

        },
        sql,
        1,
        3,
        2,
        new AbstractFunction1<ResultSet, String>() {
          @Override
          public String apply(ResultSet rs) {
            try {
              return rs.getString(1) + "," + rs.getString(2);
            } catch (SQLException e) {
              throw new RuntimeException("result set error");
            }
          }
        },
        ClassManifestFactory.fromClass(String.class)
    );
    JavaRDD<String> javaRdd = JavaRDD
        .fromRDD(jdbcRdd, ClassManifestFactory.fromClass(String.class));

    javaRdd.foreach(System.out::println);
    jsc.stop();
  }

}
